package com.diao.flink.windowAndTimer;

import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

/**
 * @author: chenzhidiao
 * @date: 2021/2/4 16:20
 * @description: 每隔3秒进行一次单词计数，使用EventTime，引入WaterMark解决乱序
 * @version: 1.0
 */
public class TumbingWindowWordCountWithWatermark {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(5);

        DataStreamSource<String> dStream = env.socketTextStream("localhost", 9999);

        dStream.map(new MapFunction<String, Tuple2<String,Long>>() {

            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                String[] fields = s.split(",");
                return new Tuple2<String,Long>(fields[0],Long.valueOf(fields[1]));
            }
            //获取数据里面的EventTime
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .forGenerator((ctx)->new PeriodicWatermarkGenerator())
                        .withTimestampAssigner(ctx -> new TimeStampExtractor())
        )
                .keyBy(tuple->tuple.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .process(new SumProcessWindowFunction())
                .print()
                .setParallelism(1);

    }

    /**
     * watermark的生成方式
     */
    private static class PeriodicWatermarkGenerator implements WatermarkGenerator<Tuple2<String, Long>> {


        //每来一条事件数据调用一次，可以检查或者记录事件的时间戳，或者也可以基于事件数据本身去生成 watermark
        @Override
        public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput watermarkOutput) {
            //System.out.println(eventTimestamp);
        }

        //周期性地效用，也许会生成新的 watermark，也许不会
        //调用此方法生成 watermark 的时间间隔由 ExecutionFonfig.getAutoWatermarkInterval()决定
        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
            watermarkOutput.emitWatermark(new Watermark(System.currentTimeMillis()-5000));

        }
    }

    /**
     * 指定EventTime字段
     */
    private static class TimeStampExtractor implements TimestampAssigner<Tuple2<String,Long>> {

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
            return element.f1;
        }
    }

    /**
     * IN：输入的数据类型
     * OUT：输出的数据类型
     * KEY：Key的数据类型
     * W extends Window：Window的数据类型
     */
    private static class SumProcessWindowFunction extends ProcessWindowFunction<Tuple2<String,Long>,String, String, TimeWindow> {

        FastDateFormat dateformat = FastDateFormat.getInstance("HH:mm:ss");

        @Override
        public void process(String key,  Context context,  Iterable<Tuple2<String, Long>> elements,  Collector<String> out) throws Exception {
            System.out.println("处理时间："+dateformat.format(context.currentProcessingTime()));
            System.out.println("window start time:"+dateformat.format(context.window().getStart()));


            ArrayList<Object> list = new ArrayList<>();
            for (Tuple2<String, Long> element : elements) {
                list.add(element.toString()+"|"+dateformat.format(element.f1));
            }
            out.collect(list.toString());
            System.out.println("window end time:"+dateformat.format(context.window().getEnd()));

        }
    }
}
